package in.co.codedoc.event;

import in.co.codedoc.cg.annotations.DBType;
import in.co.codedoc.cg.annotations.HasProperties;
import in.co.codedoc.cg.annotations.IsAnIOCComponent;
import in.co.codedoc.cg.annotations.IsMappedToJSON;
import in.co.codedoc.cg.annotations.Property;
import in.co.codedoc.cg.annotations.ShortType;
import in.co.codedoc.config.Configurable;
import in.co.codedoc.config.ConfigurationUtil;
import in.co.codedoc.config.Configurator;
import in.co.codedoc.ioc.IOCContainer;
import in.co.codedoc.ioc.Startable;
import in.co.codedoc.ioc.Stoppable;
import in.co.codedoc.json.JSONUtil;
import in.co.codedoc.json.JSONValue;
import in.co.codedoc.sql.DBInteractions;
import in.co.codedoc.sql.DBMapUtil;
import in.co.codedoc.sql.LimitClause;
import in.co.codedoc.sql.OrderingClause;
import in.co.codedoc.sql.WhereClause;
import in.co.codedoc.sql.WhereClause.Op;
import in.co.codedoc.time.DayTime;

import java.util.HashMap;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@IsAnIOCComponent
@HasProperties
({
	@Property(name="batchSize",datatype=ShortType.class,defaultValue="(short)10"),
	@Property(name="threadPoolSize",datatype=ShortType.class,defaultValue="(short)2"),
	@Property(name="numEventsBeforeRejuvination",datatype=ShortType.class,defaultValue="(short)100"),
	@Property(name="waitBetweenRuns",datatype=ShortType.class,defaultValue="(short)1000"),
	@Property(name="numDeliveryGraceSeconds",datatype=ShortType.class,defaultValue="(short)60"),
	@Property(name="unitDelay",datatype=ShortType.class,defaultValue="(short)100")
})
@IsMappedToJSON
public class AsynchronousEventProcessor
	extends AsynchronousEventProcessorBase
	implements Startable,Stoppable,Runnable,Configurable
{
	AsynchronousEventProcessor()
	{
	}
	
	public AsynchronousEventProcessor(AsynchronousEventRegistry aer,Configurator configurator)
	{
		this();
		this.aer = aer;
	}
	
	@Override
	public AsynchronousEventProcessor SetBatchSize(short batchSize)
	{
		if(batchSize > 0 && batchSize < GetThreadPoolSize())
		{
			SetThreadPoolSize(batchSize);
		}
		
		return super.SetBatchSize(batchSize);
	}

	@Override
	public AsynchronousEventProcessor SetThreadPoolSize(short threadPoolSize)
	{
		if(threadPoolSize > 0)
		{
			executor.setCorePoolSize(threadPoolSize);
			executor.setMaximumPoolSize(threadPoolSize);
		}
		
		return super.SetThreadPoolSize(threadPoolSize);
	}

	@Override
	public void Configure(JSONValue jsonData)
	{
		ConfigurationUtil.ConfigureUsingProperties(this, jsonData);
	}

	@Override
	public String GetConfigSectionName()
	{
		return "AsynchronousEventProcessor";
	}
	
	@Override
	public void Start()
	{
		if(GetBatchSize() <= 0)
		{
			Logger.logger.info("AsynchronousEventProcessor Batch size set to " + GetBatchSize() + ". Shutting down.");
			shutdown = true;
		}
		else	
		{
			shutdown = false;
			thread = new Thread(this);
			thread.start();
		}
	}

	@Override
	public void Stop()
	{	
		executor.shutdown();
	
		shutdown = true;
		if(thread != null)
		{
			thread.interrupt();
		}
		
		short numWaits = 0;
		while(!executor.isShutdown() && (numWaits < 10))
		{
			try
			{
				Thread.sleep(10000);
			}
			catch(Throwable th)
			{
				//ignore
			}
			numWaits++;
		}
		
		if(!executor.isShutdown())
		{
			executor.shutdownNow();
		}
	}

	@Override
	public void run()
	{
		while(!shutdown)
		{
			try
			{
				RunOnce();
			}
			catch(Throwable th)
			{
				Logger.logger.error
				(
					"Exception while processing asynch events(IGNORED):" + th.getMessage(),
					th
				);
			}

			try
			{
				Thread.sleep(GetWaitBetweenRuns());
			}
			catch(Throwable th)
			{
				Logger.logger.debug("Exception while waiting between runs:" + th.getMessage(),th);
			}
		}
	}
	
	private void RunOnce()
		throws Throwable
	{
		HashMap<AsyncEventMeta,AsyncEvent> events = null;
		
		DBInteractions.StartNewTransaction();

		try
		{
			events = GrabEventsForProcessing();
			DBInteractions.CommitTransaction();
		}
		catch(Throwable th)
		{
			try
			{
				DBInteractions.RollbackTransaction();
			}
			catch(Throwable th1)
			{
				Logger.logger.error
				(
					"Exception while rolling back in the context of another exception." + 
					" (Loggin the original exception rigth after this):" + th1.getMessage(),
					th1
				);
				
				Logger.logger.error
				(
					"Original Exception(that is re-raised immediately after logging this logging):" 
					+ th.getMessage(),
					th
				);
			}
			
			throw new RuntimeException(th);
		}	
			
		if(events == null || events.isEmpty())
		{
			return;
		}
		
		for(final AsyncEventMeta aem:events.keySet())
		{
			final AsyncEvent ae = events.get(aem);
			
			executor.execute
			(
				new Runnable()
				{
					@Override
					public void run()
					{
						try
						{
							DBInteractions.StartNewTransaction();
							try
							{
								ProcessEvent(aem,ae);
								DBInteractions.CommitTransaction();
							}
							catch(Throwable th)
							{
								Logger.logger.error
								(
									"PocessEvents ideally SHOULD NTO raise any exceptions. But caught:" + 
									th.getMessage(),
									th
								);
								DBInteractions.RollbackTransaction();
							}
						}
						catch(Throwable th)
						{
							Logger.logger.error
							(
								"Exception in last line of defense:" + th.getMessage(),
								th
							);
						}
					}
				}
			);
		}
	}

	private HashMap<AsyncEventMeta,AsyncEvent> GrabEventsForProcessing()
		throws Throwable
	{
		HashMap<AsyncEventMeta,AsyncEvent> events = new HashMap<AsyncEventMeta, AsyncEvent>();
		
		ResetStaleEvents();
		
		int thisBatchSize = GetBatchSize() - queue.size();
		
		if(thisBatchSize <= 0)
		{
			return events;
		}
		
		DBMapUtil mapUtil = IOCContainer.Lookup(DBMapUtil.class);
		
		List<AsyncEventMeta> batch
			= AsyncEventMeta.Find
				(
					new WhereClause()
						.And
						(
							AsyncEventMetaColumns.EarliestNextPickup,
							Op.IS_LESS_THAN_OR_EQUAL_TO,
							mapUtil.GetInputColumnMap(new DayTime(), DBType.TIMESTAMP)
						)
						.And
						(
							AsyncEventMetaColumns.PickedUpAt,
							Op.IS_NULL
						),
						new OrderingClause()
							.Descending(AsyncEventMetaColumns.EarliestNextPickup),
						new LimitClause(thisBatchSize)
				);
		
		for(final AsyncEventMeta aem:batch)
		{
			try
			{
				aem.SetPickedUpAt(new DayTime());
				aem.Save();
			}
			catch(Throwable th)
			{
				Logger.logger.debug
				(
					"Failed to grab event " + aem + ". Probably lost a conflict with some other machine. Ignoring."
				);
				continue;
			}
			
			List<AsyncEvent> aes
				= AsyncEvent.Find(new WhereClause().And(AsyncEventColumns.MetaId,aem.GetId()));
			
			if(aes.isEmpty())
			{
				Logger.logger.warn
				(
					"Event data for meta " + aem +
					" is missing. Removing it." +
					" But things look suspicious. Probably a bug."
				);
				
				aem.Delete();
			}
			else
			{
				if(aes.size() > 1)
				{
					Logger.logger.warn
					(
						"More than one AsyncEvent entry associated with meta " + aem + 
						":" + aes + ". Only the first one will be processed." +
						" Rest will be ignored and not even deleted."
					);
				}
				
				final AsyncEvent ae = aes.get(0);
			
				events.put(aem,ae);
			}
		}
		
		return events;
	}
	
	private void ResetStaleEvents()
	{
		//Grab back one batch worth at the most - distributing the grabbing
		//back "load" across runs. But this strategy looks funny because in the first
		//place there should not be too many of these: in a health system.
		
		DBMapUtil mapUtil = IOCContainer.Lookup(DBMapUtil.class);
		
		List<AsyncEventMeta> staleEvents
			= AsyncEventMeta.Find
				(
					new WhereClause()
						.And
						(
							AsyncEventMetaColumns.PickedUpAt,
							Op.IS_NOT_NULL
						).And
						(
							AsyncEventMetaColumns.PickedUpAt,
							Op.IS_LESS_THAN,
							mapUtil.GetInputColumnMap(new DayTime(-1* GetNumDeliveryGraceSeconds() * 1000), DBType.TIMESTAMP)
						),
						null,
						new LimitClause(GetBatchSize())
				);
		
		for(AsyncEventMeta staleEvent:staleEvents)
		{
			RescheduleEvent(staleEvent);
		}

	}

	private void ProcessEvent(AsyncEventMeta aem, AsyncEvent ae)
	{
		try
		{
			aer.Dispatch
			(
				JSONUtil.Decode
				(
					JSONValue.Decode(ae.GetEventData()),
					aem.GetEventType()
				)
			);
			try
			{
				ae.Delete();
				aem.Delete();
			}
			catch(Throwable th)
			{
				Logger.logger.error
				(
					"Processed events but failed to remove DB entries for " +
					aem + ":" + th.getMessage(),
					th
				);
			}
		}
		catch(Throwable th)
		{
			Logger.logger.error("Failed to process event " + aem + ":" + th.getMessage(),th);
			
			RescheduleEvent(aem);
		}		
	}

	private void RescheduleEvent(AsyncEventMeta aem)
	{
		try
		{
			aem.SetPickedUpAt(null);
			aem.SetEarliestNextPickup(GetEarliestNextPickup(aem.GetVersion()));
			aem.Save();
		}
		catch(Throwable th)
		{
			Logger.logger.error("Failed to reschedule event " + aem + ":" + th.getMessage(),th);
		}
	}

	private DayTime GetEarliestNextPickup(short version)
	{
		return new DayTime(GetUnitDelay() * ((long)(Math.pow(10,version/2))));
	}
	
	private class MyThread
		extends Thread
	{
		public MyThread(Runnable r)
		{
			super(r);
			setName("AEP-Worker-" + GetAepWorkerId());
		}
		
		public void OneMoreEvent()
		{
			numEvents++;
		}
		
		public boolean NeedsRejuvination()
		{
			return numEvents > GetNumEventsBeforeRejuvination();
		}
		
		private int numEvents = 0;
	}
	
	private class MyThreadFactory
		implements ThreadFactory
	{
		@Override
		public Thread newThread(Runnable r)
		{
			return new MyThread(r);
		}
	}
	
	private AsynchronousEventRegistry aer;
	private Thread thread;
	private boolean shutdown = false;
	
	private BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
	private ThreadPoolExecutor executor
		= new ThreadPoolExecutor(1,1,0,TimeUnit.SECONDS,queue,new MyThreadFactory())
	{
		@Override
		protected void afterExecute(Runnable r, Throwable t)
		{
			super.afterExecute(r, t);
			
			Thread currentThread = Thread.currentThread();
			if(currentThread instanceof MyThread)
			{
				
				((MyThread)currentThread).OneMoreEvent();
				
				if(((MyThread)currentThread).NeedsRejuvination())
				{
					((MyThread)currentThread).interrupt();
				}
			}
			else
			{
				Logger.logger.warn
				(
					"Some class other than MyThread has been used." +
					" This might cause some defects when the system" + 
					" is run for long periods of time."
				);
			}
		}
	};

	private static int GetAepWorkerId()
	{
		synchronized(AsynchronousEventProcessor.class)
		{
			return aepWorkerId++;
		}
	}
	
	private static int aepWorkerId = 0;
}
